package cc.xpcas.raft.role;

import java.util.concurrent.*;
import java.util.concurrent.atomic.*;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import cc.xpcas.raft.LogEntry;
import cc.xpcas.raft.context.Context;
import cc.xpcas.raft.message.*;

/**
 * @author xp
 */
public class Follower extends Role {

    private static final Logger LOG = LoggerFactory.getLogger(Follower.class);

    private final AtomicBoolean timeoutFlag = new AtomicBoolean(false);

    private Future future;

    public Follower(Context context) {
        super(context);
    }

    /**
     * in mainline
     */
    @Override
    public void start() {
        LOG.info(String.format("%s: follower start", getConfiguration().getServerId()));

        resumeHeartbeatCountdown();
    }

    /**
     * in mainline
     */
    @Override
    public void end() throws InterruptedException {
        stopHeartbeatCountdown(true);
    }

    /**
     * in mainline
     */
    private boolean stopHeartbeatCountdown(boolean force) throws InterruptedException {
        if (!force) {
            if (!timeoutFlag.compareAndSet(false, true)) {
                // already timeout
                return false;
            }
        } else {
            timeoutFlag.set(true);
        }
        if (future != null) {
            future.cancel(false);
        }
        return true;
    }

    /**
     * in mainline
     */
    private void resumeHeartbeatCountdown() {
        timeoutFlag.set(false);
        future = getContext().schedule(this::heartbeatTimeoutHandler, getConfiguration().getHeartbeatTimeout());
    }

    /**
     * in mainline
     */
    private void resetHeartbeatCountdown() throws InterruptedException {
        if (stopHeartbeatCountdown(false)) {
            resumeHeartbeatCountdown();
        }
    }

    private boolean isValidLeader(long term) {
        /*
         * -1 new leader
         *  0 current leader
         *  1 old leader
         */
        return getContext().catchUpTerm(term) <= 0;
    }

    @Override
    public boolean handleAppendEntriesRequest(MessageContext messageContext) throws InterruptedException {
        LOG.info(String.format("%s: handle AppendEntriesRequest",
                getConfiguration().getServerId()
        ));

        AppendEntriesRequest message = messageContext.getMessage();
        if (isValidLeader(message.getTerm())) {
            if (stopHeartbeatCountdown(false)) {
                try {
                    boolean append = true;
                    if (message.getPreviousLogId() != 0) {
                        LogEntry existed = getContext().getLogEntry(message.getPreviousLogId());
                        if (existed == null || existed.getTerm() != message.getPreviousLogTerm()) {
                            // previous log doesn't match
                            append = false;
                        }
                    }
                    if (append) {
                        getContext().appendLogEntries(message.getEntries());
                        getContext().catchUpCommittedLogId(message.getLastCommittedLogId());
                    }

                    AppendEntriesResponse response = new AppendEntriesResponse();
                    response.setServerId(getConfiguration().getServerId());
                    response.setCurrentTerm(getContext().getTerm());
                    response.setLastLogId(getContext().getLastLogId());
                    response.setLastCommittedLogId(getContext().getLastCommittedLogId());
                    messageContext.reply(response);
                } finally {
                    resumeHeartbeatCountdown();
                }
            } else {
                // already timeout
                // do nth.
            }
        } else {
            // the remote is behind
            AppendEntriesResponse response = new AppendEntriesResponse();
            response.setCurrentTerm(getContext().getTerm());
            messageContext.reply(response);
        }
        return true;
    }

    /**
     * in mainline
     */
    @Override
    public boolean handleRequestVoteRequest(MessageContext messageContext) {
        LOG.info(String.format("%s: handle RequestVoteRequest",
                getConfiguration().getServerId()
        ));

        RequestVoteRequest message = messageContext.getMessage();
        boolean granted = grantVote(message.getTerm(), message.getServerId());

        RequestVoteResponse response = new RequestVoteResponse();
        response.setCurrentTerm(getContext().getTerm());
        response.setVoteGranted(granted);
        messageContext.reply(response);

        return true;
    }

    /**
     * in mainline
     */
    private boolean grantVote(long candidateTerm, int candidateId) {
        int v = getContext().catchUpTerm(candidateTerm);
        if (v > 0) return false;
        if (v < 0 || getContext().getVotedFor() == Context.VOTED_FOR_NOBODY) {
            getContext().setVotedFor(candidateId);
        }
        return getContext().getVotedFor() == candidateId;
    }

    /**
     * in scheduled
     */
    private void heartbeatTimeoutHandler() {
        if (!timeoutFlag.compareAndSet(false, true)) {
            // already reset
            return;
        }
        getContext().becomeCandidate();
    }
}
